﻿using System;
using System.Collections.Generic;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

using StackExchange.Redis;

namespace RedisLib
{
    public partial class RedisClient
    {
        public async Task<long> StreamAcknowledge(string key, string groupName, string messageId, int db = -1, CancellationToken cancellationToken = default)
        {
            IDatabase database = await GetConnectedDatabase(db, cancellationToken);
            if (database == null) return -1;

            RedisKey redisKey = ConvertStringKeyToRedisKey(key);
            RedisValue redisGroupName = ConvertObjectValueToRedisValue(groupName);
            RedisValue redisMessageId = ConvertObjectValueToRedisValue(messageId);

            return await database.StreamAcknowledgeAsync(redisKey, redisGroupName, redisMessageId);
        }

        public async Task<long> StreamAcknowledge(string key, string groupName, List<string> messageIds, int db = -1, CancellationToken cancellationToken = default)
        {
            IDatabase database = await GetConnectedDatabase(db, cancellationToken);
            if (database == null) return -1;

            RedisKey redisKey = ConvertStringKeyToRedisKey(key);
            RedisValue redisGroupName = ConvertObjectValueToRedisValue(groupName);
            RedisValue[] redisMessageIds = BatchConvertObjectValueToRedisValue(messageIds);

            return await database.StreamAcknowledgeAsync(redisKey, redisGroupName, redisMessageIds);
        }

        public async Task<string> StreamAddAsync<T>(string key, string streamField, T streamValue, string messageId = null, int? maxLength = null, bool useApproximateMaxLength = false, int db = -1, CancellationToken cancellationToken = default)
        {
            IDatabase database = await GetConnectedDatabase(db, cancellationToken);
            if (database == null) return default;

            RedisKey redisKey = ConvertStringKeyToRedisKey(key);
            RedisValue redisStreamField = ConvertObjectValueToRedisValue(streamField);
            RedisValue redisStreamValue = ConvertObjectValueToRedisValue(streamValue);
            RedisValue? redisMessageId = default;
            if (string.IsNullOrWhiteSpace(messageId) == false)
                redisMessageId = ConvertObjectValueToRedisValue(messageId);

            RedisValue newMessageId = await database.StreamAddAsync(redisKey, redisStreamField, redisStreamValue, redisMessageId, maxLength, useApproximateMaxLength);

            return ConvertRedisValueToObjectValue<string>(newMessageId);
        }

        public async Task<string> StreamAddAsync<T>(string key, Dictionary<string, T> streamValues, string messageId = null, int? maxLength = null, bool useApproximateMaxLength = false, int db = -1, CancellationToken cancellationToken = default)
        {
            IDatabase database = await GetConnectedDatabase(db, cancellationToken);
            if (database == null) return default;

            RedisKey redisKey = ConvertStringKeyToRedisKey(key);

            NameValueEntry[] streamPairs = new NameValueEntry[streamValues.Count];
            int i = 0;
            foreach (var streamValue in streamValues)
            {
                RedisValue name = ConvertObjectValueToRedisValue(streamValue.Key);
                RedisValue value = ConvertObjectValueToRedisValue(streamValue.Value);
                streamPairs[++i] = new NameValueEntry(name, value);
            }

            RedisValue? redisMessageId = default;
            if (string.IsNullOrWhiteSpace(messageId) == false)
                redisMessageId = ConvertObjectValueToRedisValue(messageId);

            RedisValue newMessageId = await database.StreamAddAsync(redisKey, streamPairs, redisMessageId, maxLength, useApproximateMaxLength);

            return ConvertRedisValueToObjectValue<string>(newMessageId);
        }

        public async Task<(string Id, Dictionary<string, T> Values)?[]> StreamClaim<T>(string key, string consumerGroup, string claimingConsumer, long minIdleTimeInMs, List<string> messageIds, int db = -1, CancellationToken cancellationToken = default)
            where T : class
        {
            IDatabase database = await GetConnectedDatabase(db, cancellationToken);
            if (database == null) return default;

            RedisKey redisKey = ConvertStringKeyToRedisKey(key);
            RedisValue redisConsumerGroup = ConvertObjectValueToRedisValue(consumerGroup);
            RedisValue redisClaimingConsumer = ConvertObjectValueToRedisValue(claimingConsumer);
            RedisValue[] redisMessageIds = BatchConvertObjectValueToRedisValue(messageIds);

            StreamEntry[] streamEntries = await database.StreamClaimAsync(redisKey, redisConsumerGroup, redisClaimingConsumer, minIdleTimeInMs, redisMessageIds);

            (string Id, Dictionary<string, T> Values)?[] returnValues = new (string Id, Dictionary<string, T> Values)?[streamEntries.Length];
            for (int i = 0; i < streamEntries.Length; i++)
            {
                StreamEntry streamEntry = streamEntries[i];
                if (streamEntry.IsNull == false)
                {
                    Dictionary<string, T> values = [];
                    foreach (var value in streamEntry.Values)
                    {
                        values[ConvertRedisValueToObjectValue<string>(value.Name)] = ConvertRedisValueToObjectValue<T>(value.Value);
                    }

                    string id = ConvertRedisValueToObjectValue<string>(streamEntry.Id);
                    returnValues[i] = (id, values);
                }
            }

            return returnValues;
        }

        public async Task<(string Id, Dictionary<string, T?> Values)?[]> StreamClaimForValueType<T>(string key, string consumerGroup, string claimingConsumer, long minIdleTimeInMs, List<string> messageIds, int db = -1, CancellationToken cancellationToken = default)
            where T : struct
        {
            IDatabase database = await GetConnectedDatabase(db, cancellationToken);
            if (database == null) return default;

            RedisKey redisKey = ConvertStringKeyToRedisKey(key);
            RedisValue redisConsumerGroup = ConvertObjectValueToRedisValue(consumerGroup);
            RedisValue redisClaimingConsumer = ConvertObjectValueToRedisValue(claimingConsumer);
            RedisValue[] redisMessageIds = BatchConvertObjectValueToRedisValue(messageIds);

            StreamEntry[] streamEntries = await database.StreamClaimAsync(redisKey, redisConsumerGroup, redisClaimingConsumer, minIdleTimeInMs, redisMessageIds);

            (string Id, Dictionary<string, T?> Values)?[] returnValues = new (string Id, Dictionary<string, T?> Values)?[streamEntries.Length];
            for (int i = 0; i < streamEntries.Length; i++)
            {
                StreamEntry streamEntry = streamEntries[i];
                if (streamEntry.IsNull == false)
                {
                    Dictionary<string, T?> values = [];
                    foreach (var value in streamEntry.Values)
                    {
                        values[ConvertRedisValueToObjectValue<string>(value.Name)] = ConvertRedisValueToValueTypeValue<T>(value.Value);
                    }

                    string id = ConvertRedisValueToObjectValue<string>(streamEntry.Id);
                    returnValues[i] = (id, values);
                }
            }

            return returnValues;
        }

        public async Task<string[]> StreamClaimIdsOnly(string key, string consumerGroup, string claimingConsumer, long minIdleTimeInMs, List<string> messageIds, int db = -1, CancellationToken cancellationToken = default)
        {
            IDatabase database = await GetConnectedDatabase(db, cancellationToken);
            if (database == null) return [];

            RedisKey redisKey = ConvertStringKeyToRedisKey(key);
            RedisValue redisConsumerGroup = ConvertObjectValueToRedisValue(consumerGroup);
            RedisValue redisClaimingConsumer = ConvertObjectValueToRedisValue(claimingConsumer);
            RedisValue[] redisMessageIds = BatchConvertObjectValueToRedisValue(messageIds);

            RedisValue[] resultMessageIds = await database.StreamClaimIdsOnlyAsync(redisKey, redisConsumerGroup, redisClaimingConsumer, minIdleTimeInMs, redisMessageIds);

            string[] strMessageIds = new string[resultMessageIds.Length];
            for (int i = 0; i < resultMessageIds.Length; i++)
            {
                strMessageIds[i] = ConvertRedisValueToObjectValue<string>(resultMessageIds[i]);
            }

            return strMessageIds;
        }

        public async Task<bool> StreamConsumerGroupSetPosition(string key, string groupName, string position, int db = -1, CancellationToken cancellationToken = default)
        {
            IDatabase database = await GetConnectedDatabase(db, cancellationToken);
            if (database == null) return false;

            RedisKey redisKey = ConvertStringKeyToRedisKey(key);
            RedisValue redisGroupName = ConvertObjectValueToRedisValue(groupName);
            RedisValue redisPosition = ConvertObjectValueToRedisValue(position);

            return await database.StreamConsumerGroupSetPositionAsync(redisKey, redisGroupName, redisPosition);
        }

        public async Task<(string Name, int PendingMessageCount, long IdleTimeInMilliseconds)[]> StreamConsumerInfo(string key, string groupName, int db = -1, CancellationToken cancellationToken = default)
        {
            IDatabase database = await GetConnectedDatabase(db, cancellationToken);
            if (database == null) return default;

            RedisKey redisKey = ConvertStringKeyToRedisKey(key);
            RedisValue redisGroupName = ConvertObjectValueToRedisValue(groupName);

            StreamConsumerInfo[] streamConsumerInfos = await database.StreamConsumerInfoAsync(redisKey, redisGroupName);
            (string Name, int PendingMessageCount, long IdleTimeInMilliseconds)[] results = new (string Name, int PendingMessageCount, long IdleTimeInMilliseconds)[streamConsumerInfos.Length];
            for (int i = 0; i < streamConsumerInfos.Length; i++)
            {
                var streamConsumerInfo = streamConsumerInfos[i];
                results[i] = (streamConsumerInfo.Name, streamConsumerInfo.PendingMessageCount, streamConsumerInfo.IdleTimeInMilliseconds);
            }

            return results;
        }

        public async Task<bool> StreamCreateConsumerGroup(string key, string groupName, string position, bool createStream = true, int db = -1, CancellationToken cancellationToken = default)
        {
            IDatabase database = await GetConnectedDatabase(db, cancellationToken);
            if (database == null) return false;

            RedisKey redisKey = ConvertStringKeyToRedisKey(key);
            RedisValue redisGroupName = ConvertObjectValueToRedisValue(groupName);
            RedisValue? redisPosition = null;
            if (string.IsNullOrWhiteSpace(position) == false) redisPosition = ConvertObjectValueToRedisValue(position);

            return await database.StreamCreateConsumerGroupAsync(redisKey, redisGroupName, redisPosition, createStream);
        }

        public async Task<long> StreamDelete(string key, List<string> messageIds, int db = -1, CancellationToken cancellationToken = default)
        {
            IDatabase database = await GetConnectedDatabase(db, cancellationToken);
            if (database == null) return -1;

            RedisKey redisKey = ConvertStringKeyToRedisKey(key);
            RedisValue[] redisMessageIds = BatchConvertObjectValueToRedisValue(messageIds);

            return await database.StreamDeleteAsync(redisKey, redisMessageIds);
        }

        public async Task<long> StreamDeleteConsumer(string key, string groupName, string consumerName, int db = -1, CancellationToken cancellationToken = default)
        {
            IDatabase database = await GetConnectedDatabase(db, cancellationToken);
            if (database == null) return -1;

            RedisKey redisKey = ConvertStringKeyToRedisKey(key);
            RedisValue redisGroupName = ConvertObjectValueToRedisValue(groupName);
            RedisValue redisConsumerName = ConvertObjectValueToRedisValue(consumerName);

            return await database.StreamDeleteConsumerAsync(redisKey, redisGroupName, redisConsumerName);
        }

        public async Task<bool> StreamDeleteConsumerGroup(string key, string groupName, int db = -1, CancellationToken cancellationToken = default)
        {
            IDatabase database = await GetConnectedDatabase(db, cancellationToken);
            if (database == null) return false;

            RedisKey redisKey = ConvertStringKeyToRedisKey(key);
            RedisValue redisGroupName = ConvertObjectValueToRedisValue(groupName);

            return await database.StreamDeleteConsumerGroupAsync(redisKey, redisGroupName);
        }

        public async Task<(string Name, int ConsumerCount, int PendingMessageCount, string LastDeliveredId)[]> StreamGroupInfo(string key, int db = -1, CancellationToken cancellationToken = default)
        {
            IDatabase database = await GetConnectedDatabase(db, cancellationToken);
            if (database == null) return default;

            RedisKey redisKey = ConvertStringKeyToRedisKey(key);
            StreamGroupInfo[] streamGroupInfos = await database.StreamGroupInfoAsync(redisKey);

            (string Name, int ConsumerCount, int PendingMessageCount, string LastDeliveredId)[] results = new (string Name, int ConsumerCount, int PendingMessageCount, string LastDeliveredId)[streamGroupInfos.Length];
            for (int i = 0; i < streamGroupInfos.Length; i++)
            {
                var streamGroupInfo = streamGroupInfos[i];
                results[i] = (streamGroupInfo.Name, streamGroupInfo.ConsumerCount, streamGroupInfo.PendingMessageCount, streamGroupInfo.LastDeliveredId);
            }

            return results;
        }

        public async Task<StreamInfo> StreamInfo(string key, int db = -1, CancellationToken cancellationToken = default)
        {
            IDatabase database = await GetConnectedDatabase(db, cancellationToken);
            if (database == null) return default;

            RedisKey redisKey = ConvertStringKeyToRedisKey(key);
            return await database.StreamInfoAsync(redisKey);
        }

        public async Task<long> StreamLength(string key, int db = -1, CancellationToken cancellationToken = default)
        {
            IDatabase database = await GetConnectedDatabase(db, cancellationToken);
            if (database == null) return -1;

            RedisKey redisKey = ConvertStringKeyToRedisKey(key);
            return await database.StreamLengthAsync(redisKey);
        }

        public async Task<StreamPendingInfo> StreamPending(string key, string groupName, int db = -1, CancellationToken cancellationToken = default)
        {
            IDatabase database = await GetConnectedDatabase(db, cancellationToken);
            if (database == null) return default;

            RedisKey redisKey = ConvertStringKeyToRedisKey(key);
            RedisValue redisGroupName = ConvertObjectValueToRedisValue(groupName);

            return await database.StreamPendingAsync(redisKey, redisGroupName);
        }

        public async Task<StreamPendingMessageInfo[]> StreamPendingMessages(string key, string groupName, int count, string consumerName, string minId = null, string maxId = null, int db = -1, CancellationToken cancellationToken = default)
        {
            IDatabase database = await GetConnectedDatabase(db, cancellationToken);
            if (database == null) return default;

            RedisKey redisKey = ConvertStringKeyToRedisKey(key);
            RedisValue redisGroupName = ConvertObjectValueToRedisValue(groupName);
            RedisValue redisConsumerName = ConvertObjectValueToRedisValue(consumerName);

            RedisValue? redisMinId = null;
            if (string.IsNullOrWhiteSpace(minId) == false) redisMinId = ConvertObjectValueToRedisValue(minId);

            RedisValue? redisMaxId = null;
            if (string.IsNullOrWhiteSpace(maxId) == false) redisMaxId = ConvertObjectValueToRedisValue(maxId);

            return await database.StreamPendingMessagesAsync(redisKey, redisGroupName, count, redisConsumerName, redisMinId, redisMaxId);
        }

        public async Task<StreamEntry[]> StreamRange(string key, string minId = null, string maxId = null, int? count = null, Order messageOrder = Order.Ascending, int db = -1, CancellationToken cancellationToken = default)
        {
            IDatabase database = await GetConnectedDatabase(db, cancellationToken);
            if (database == null) return default;

            RedisKey redisKey = ConvertStringKeyToRedisKey(key);

            RedisValue? redisMinId = null;
            if (string.IsNullOrWhiteSpace(minId) == false) redisMinId = ConvertObjectValueToRedisValue(minId);

            RedisValue? redisMaxId = null;
            if (string.IsNullOrWhiteSpace(maxId) == false) redisMaxId = ConvertObjectValueToRedisValue(maxId);

            return await database.StreamRangeAsync(redisKey, redisMinId, redisMaxId, count, messageOrder);
        }

        public async Task<StreamEntry[]> StreamRead(string key, string position, int? count = null, int db = -1, CancellationToken cancellationToken = default)
        {
            IDatabase database = await GetConnectedDatabase(db, cancellationToken);
            if (database == null) return default;

            RedisKey redisKey = ConvertStringKeyToRedisKey(key);
            RedisValue redisPosition = ConvertObjectValueToRedisValue(position);

            return await database.StreamReadAsync(redisKey, redisPosition, count);
        }

        public async Task<StreamEntry[]> StreamReadGroup(string key, string groupName, string consumerName, string position, int? count, bool noAck = false, int db = -1, CancellationToken cancellationToken = default)
        {
            IDatabase database = await GetConnectedDatabase(db, cancellationToken);
            if (database == null) return default;

            RedisKey redisKey = ConvertStringKeyToRedisKey(key);
            RedisValue redisGroupName = ConvertObjectValueToRedisValue(groupName);
            RedisValue redisConsumerName = ConvertObjectValueToRedisValue(consumerName);
            RedisValue? redisPosition = null;
            if (string.IsNullOrWhiteSpace(position) == false) redisPosition = ConvertObjectValueToRedisValue(position);

            return await database.StreamReadGroupAsync(redisKey, redisGroupName, redisConsumerName, redisPosition, count, noAck);
        }

        public async Task<long> StreamTrim(string key, int maxLength, bool useApproximateMaxLength = false, int db = -1, CancellationToken cancellationToken = default)
        {
            IDatabase database = await GetConnectedDatabase(db, cancellationToken);
            if (database == null) return -1;

            RedisKey redisKey = ConvertStringKeyToRedisKey(key);
            return await database.StreamTrimAsync(redisKey, maxLength, useApproximateMaxLength);
        }
    }
}
